package com.example.module.iot.aws.config;

import cn.hutool.core.util.ObjectUtil;
import com.example.module.iot.aws.callback.AwsMqttConnectionCallback;
import com.example.module.iot.aws.service.AwsMqttMessageConsumer;
import com.example.module.iot.aws.util.AwsMqttUtils;
import com.example.module.iot.constant.Constants;
import jakarta.annotation.PreDestroy;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import software.amazon.awssdk.crt.CrtResource;
import software.amazon.awssdk.crt.http.HttpProxyOptions;
import software.amazon.awssdk.crt.mqtt.MqttClientConnection;
import software.amazon.awssdk.crt.mqtt.MqttClientConnectionEvents;
import software.amazon.awssdk.crt.mqtt.MqttMessage;
import software.amazon.awssdk.crt.mqtt.QualityOfService;
import software.amazon.awssdk.iot.AwsIotMqttConnectionBuilder;

import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;

@Slf4j
@Configuration
@ConditionalOnProperty(name = Constants.MQTT_AWS_ENABLED) // 默认关闭
@EnableConfigurationProperties(AwsMqttProperties.class)
public class AwsMqttConfiguration {

    private MqttClientConnection mqttClientConnection;

    @Bean
    public MqttClientConnectionEvents mqttClientConnectionEvents() {
        return new AwsMqttConnectionCallback();
    }

    @Bean
    public Consumer<MqttMessage> consumer() {
        return new AwsMqttMessageConsumer();
    }

    @SneakyThrows
    @Bean
    public MqttClientConnection mqttClientConnection(AwsMqttProperties awsMqttProperties) {
        AwsIotMqttConnectionBuilder builder = AwsIotMqttConnectionBuilder.newMtlsBuilderFromPath(awsMqttProperties.getCert(), awsMqttProperties.getKey());
        if (ObjectUtil.isNotEmpty(awsMqttProperties.getCa())) {
            builder.withCertificateAuthorityFromPath(null, awsMqttProperties.getCa());
        }
        builder.withConnectionEventCallbacks(mqttClientConnectionEvents())
                .withClientId(awsMqttProperties.getClientId())
                .withEndpoint(awsMqttProperties.getEndpoint())
                .withPort(awsMqttProperties.getPort())
                .withCleanSession(true)
                .withProtocolOperationTimeoutMs(60000);
        if (ObjectUtil.isNotEmpty(awsMqttProperties.getProxyHost()) && awsMqttProperties.getProxyPort() > 0) {
            HttpProxyOptions proxyOptions = new HttpProxyOptions();
            proxyOptions.setHost(awsMqttProperties.getProxyHost());
            proxyOptions.setPort(awsMqttProperties.getProxyPort());
            builder.withHttpProxyOptions(proxyOptions);
        }
        MqttClientConnection connection = builder.build();
        builder.close();

        // Connect the MQTT client
        CompletableFuture<Boolean> connected = connection.connect();
        try {
            boolean sessionPresent = connected.get();
            log.info("Connected to {} session!", !sessionPresent ? "new" : "existing");
        } catch (Exception ex) {
            throw new RuntimeException("Exception occurred during connect", ex);
        }

        connection.subscribe(awsMqttProperties.getTopic(), QualityOfService.AT_LEAST_ONCE, consumer()).get();

        this.mqttClientConnection = connection;
        return connection;
    }

    @Bean
    public AwsMqttUtils awsMqttUtils() {
        return new AwsMqttUtils(mqttClientConnection);
    }

    @SneakyThrows
    @PreDestroy
    private void destroy() { // Lifecycle annotation requires a no-arg method
        log.debug("==== PreDestroy ====");
        if (Objects.nonNull(mqttClientConnection)) {
            // Disconnect
            CompletableFuture<Void> disconnected = mqttClientConnection.disconnect();
            disconnected.get();

            // Close the connection now that we are completely done with it.
            mqttClientConnection.close();

            CrtResource.waitForNoResources();
            log.debug("Close AWS MQTT Complete!");
        }
    }

}
